package com.atguigu.dga.governance.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.assessor.Assessor;
import com.atguigu.dga.dolphinscheduler.bean.TDsTaskDefinition;
import com.atguigu.dga.dolphinscheduler.bean.TDsTaskInstance;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskDefinitionService;
import com.atguigu.dga.dolphinscheduler.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.mapper.GovernanceMetricMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.mapper.TableMetaInfoMapper;
import com.atguigu.dga.util.SpringBeanProvider;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.tools.ant.types.resources.Last;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.apache.calcite.linq4j.EnumerableDefaults.last;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author liuxu
 * @since 2023-10-11
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {

    @Autowired
    SpringBeanProvider springBeanProvider;

    @Autowired
    TableMetaInfoMapper tableMetaInfoMapper;

    @Autowired
    GovernanceMetricService governanceMetricService;

    @Autowired
    GovernanceAssessDetailMapper governanceAssessDetailMapper;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;

    //主考评方法
    @Override
    public void mainAssess(String assessDate) {

        //幂等处理
        remove(
                new QueryWrapper<GovernanceAssessDetail>().eq("assess_date",assessDate)
        );
        //读取待考评的所有表
        List<TableMetaInfo> tableMetaInfos = tableMetaInfoMapper.selectTableMetaInfoList(assessDate);
        HashMap<String, TableMetaInfo> stringTableMetaInfoHashMap = new HashMap<>();
        for (TableMetaInfo tableMetaInfo : tableMetaInfos) {
            stringTableMetaInfoHashMap.put(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName(),tableMetaInfo);
        }

        //读取所有启用的指标
        List<GovernanceMetric> governanceMetricList = governanceMetricService.list(
                new QueryWrapper<GovernanceMetric>()
                        .eq("is_disabled", "0")
        );

        //创建用于保存考评结果对象的集合
        //ArrayList<GovernanceAssessDetail> governanceAssessDetails =
                //new ArrayList<>(tableMetaInfos.size()*governanceMetricList.size());

        //处理ds的相关内容
        //从DS中查询所有的任务定义,并保存到map集合中
        List<TDsTaskDefinition> tDsTaskDefinitions = tDsTaskDefinitionService.selecList();
        HashMap<String, TDsTaskDefinition> stringTDsTaskDefinitionHashMap = new HashMap<>();
        for (TDsTaskDefinition tDsTaskDefinition : tDsTaskDefinitions) {
            stringTDsTaskDefinitionHashMap.put(tDsTaskDefinition.getName(),tDsTaskDefinition);
        }


        //从DS中查询所有的任务实列,并保存到map集合中
        List<TDsTaskInstance> tDsTaskInstances = tDsTaskInstanceService.selectList(assessDate);
        HashMap<String, TDsTaskInstance> stringTDsTaskInstanceHashMap = new HashMap<>();
        for (TDsTaskInstance tDsTaskInstance : tDsTaskInstances) {
            stringTDsTaskInstanceHashMap.put(tDsTaskInstance.getName(),tDsTaskInstance);
        }
        //创建线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 15, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2000));

        //创建任务集合
        ArrayList<CompletableFuture<GovernanceAssessDetail>> arrayList = new ArrayList<>(tableMetaInfos.size()*governanceMetricList.size());

        //每张表每个指标双层循环开始逐个考评
        for (TableMetaInfo tableMetaInfo : tableMetaInfos) {
            for (GovernanceMetric governanceMetric : governanceMetricList) {

                //处理白名单
                String skipAssessTables = governanceMetric.getSkipAssessTables();
                boolean isAssess=true;
                if (skipAssessTables != null && !skipAssessTables.trim().isEmpty()){
                    String[] split = skipAssessTables.split(",");
                    for (String s : split) {
                        if(s.equals(tableMetaInfo.getTableName())){
                            isAssess=false;
                            break;
                        }
                    }
                }
                if (!isAssess){
                    break;
                }

                //获取考评器对象
                //获取指标编码
                String metricCode = governanceMetric.getMetricCode();
                //获取考评器对象
                Assessor bean = springBeanProvider.getBean(metricCode, Assessor.class);

                //封装考评对象参数
                AssessParam assessParam = new AssessParam();
                assessParam.setAssessDate(assessDate);
                assessParam.setGovernanceMetric(governanceMetric);
                assessParam.setTableMetaInfo(tableMetaInfo);
                assessParam.setTableMetaInfoMaps(stringTableMetaInfoHashMap);
                assessParam.setTDsTaskInstance(stringTDsTaskInstanceHashMap.get(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName()));
                assessParam.setTDsTaskDefinition(stringTDsTaskDefinitionHashMap.get(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName()));

                //调用考评器对象开始考评
                //GovernanceAssessDetail governanceAssessDetail = bean.doAssess(assessParam);

                //将考评结果加入集合
                //governanceAssessDetails.add(governanceAssessDetail);

                //编排future
                CompletableFuture<GovernanceAssessDetail> governanceAssessDetailCompletableFuture = CompletableFuture.supplyAsync(
                        () -> {
                            return bean.doAssess(assessParam);
                        },
                        threadPoolExecutor
                );
                //将线程任务加入任务列表
                arrayList.add(governanceAssessDetailCompletableFuture);
            }
        }
        //线程任务执行
        List<GovernanceAssessDetail> collect = arrayList.stream().map(CompletableFuture::join).collect(Collectors.toList());
        //批量把考评结果写入表中
        saveBatch(collect);

    }

    @Override
    public String getProblemListService(String governType, Integer pageNo, Integer pageSize) {
        Integer startNum=(pageNo-1)*pageSize;
        List<GovernanceAssessDetail> governanceAssessDetailList= list(
                new QueryWrapper<GovernanceAssessDetail>().
                        eq("governance_type",governType).
                        lt("assess_score",10).
                        inSql("assess_date","select max(assess_date) from governance_assess_detail").
                        last("limit "+ startNum+" , "+pageSize)
                );

        return JSON.toJSONString(governanceAssessDetailList);
    }

    @Override
    public String getProblemNumService() {
        List<Map<String, Object>> maps = listMaps(
                new QueryWrapper<GovernanceAssessDetail>()
                        .select("governance_type", "count(*) assess_num")
                        .lt("assess_score",10)
                        .inSql("assess_date", "select max(assess_date) from governance_assess_detail")
                        .groupBy("governance_type")
        );
        HashMap<Object, Object> stringObjectHashMap = new HashMap<>();
        for (Map<String, Object> map : maps) {
            stringObjectHashMap.put(map.get("governance_type"),map.get("assess_num"));
        }

        return JSONObject.toJSONString(stringObjectHashMap);
    }
}
